package com.itheima.utils;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson2.JSON;
import com.itheima.dto.WebsocketDto;
import com.itheima.enums.WebsocketEnum;
import com.itheima.service.NewsService;
import com.itheima.service.NoticeService;
import com.itheima.vo.WebsocketVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.stream.Collectors;


/**
 * <h1>Websocket类</h1>
 *
 * @author <a href="515857087@qq.com">叶世铭</a>
 * @since 2022/10/27
 */
@Component
@Slf4j
@ServerEndpoint("/websocket/{userId}")  // 接口路径 ws://localhost:8081/websocket/userId;
public class Websocket {

    //与某个客户端的连接会话，需要通过它来给客户端发送数据
    private Session session;
    /**
     * 用户ID
     */
    private String userId;                         
    private static Websocket webSocket;
    @Resource
    private NewsService newsService;
    @Resource
    private NoticeService noticeService;

    @PostConstruct
    public void init() {
        webSocket = this;
        webSocket.newsService = this.newsService;
        webSocket.noticeService = this.noticeService;
    }

    //concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
    //虽然@Component默认是单例模式的，但springboot还是会为每个websocket连接初始化一个bean，所以可以用一个静态set保存起来。
    //  注：底下WebSocket是当前类名
    private static CopyOnWriteArraySet<Websocket> websockets = new CopyOnWriteArraySet<>();
    // 用来存在线连接用户信息
    private static ConcurrentHashMap<String, Session> sessionPool = new ConcurrentHashMap<>();
    // 公用存储消息，用来存放未读的消息
    private static ConcurrentHashMap<String, Map<String, List<String>>> messageMap = new ConcurrentHashMap<>();


    /**
     * 链接成功调用的方法,userId为新加入websocket的用户id
     */
    @OnOpen
    public void onOpen(Session session, @PathParam(value = "userId") String userId) throws Exception {
        if (sessionPool.containsKey(userId)) {
            return;
        }
        try {
            this.session = session;
            this.userId = userId;
            websockets.add(this);
            sessionPool.put(userId, session);
            log.info("【websocket消息】有新的连接，总数为:" + websockets.size() + "; 新加入的用户为：" + this.userId);
            List<String> collect = websockets.stream()
                    .map(k -> k.userId)
                    .collect(Collectors.toList());
            log.info("【当前连接的用户有】" + collect);
            offlineNotice(userId);
            offlineNews(userId);
        } catch (Exception e) {
            WebsocketVo websocketVo = new WebsocketVo();
            websocketVo.setSendUserId(CheckUtil.checkLogin());
            websocketVo.setException("建立链接异常");
            websocketVo.setType(WebsocketEnum.EXCEPTION.name());
        }
    }

    /**
     * 链接关闭调用的方法
     */
    @OnClose
    public void onClose() throws Exception {
        try {
            websockets.remove(this);
            sessionPool.remove(this.userId);
            log.info("【websocket消息】连接断开，总数为:" + websockets.size());
        } catch (Exception e) {
            WebsocketVo websocketVo = new WebsocketVo();
            websocketVo.setSendUserId(CheckUtil.checkLogin());
            websocketVo.setException("链接关闭异常");
            websocketVo.setType(WebsocketEnum.EXCEPTION.name());
        }
    }

    /**
     * 收到客户端消息后调用的方法
     * 接收用户id放在message中
     *
     * @param message 信息
     */
    @OnMessage
    public void onMessage(String message) throws Exception {
        WebsocketDto websocketDto = JSON.parseObject(message, WebsocketDto.class);
        webSocket.newsService.sendPersonalNews(websocketDto);
        log.info("【websocket消息】收到客户端消息:" + websocketDto.getContent());
    }

    /**
     * 发送错误时的处理
     *
     * @param error 错误信息
     */
    @OnError
    public void onError(Throwable error) {
        log.error("用户:" + userId + "错误,原因:" + error.getMessage());
        error.printStackTrace();
    }


    /**
     * 此为广播消息
     *
     * @param websocketVo webSocket服务端到客户端vo
     */
    public void sendAllMessage(WebsocketVo websocketVo) {
        log.info("【websocket消息】广播消息:" + websocketVo);
        for (Websocket webSocket : websockets) {
            try {
                if (webSocket.session.isOpen()) {
                    webSocket.session.getAsyncRemote().sendText(JSONUtil.toJsonStr(websocketVo));
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 此为单点消息(给发送者返回错误信息)
     *
     * @param websocketVo webSocket服务端到客户端vo
     */
    public void sendOneMessage(WebsocketVo websocketVo) {
        Session session = sessionPool.get(websocketVo.getSendUserId());
        if (session != null && session.isOpen()) {
            try {
                log.info("【websocket消息】 单点消息:" + websocketVo.getContent());
                session.getAsyncRemote().sendText(JSONUtil.toJsonStr(websocketVo));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 用于上线提示(消息)
     *
     * @param text   数据json转String
     * @param userId 用户id
     */
    public void sendOneNewsMessage(String text, String userId) {
        Session session = sessionPool.get(userId);
        if (session != null && session.isOpen()) {
            try {
                log.info("【websocket消息】 单点消息:" + text);
                session.getAsyncRemote().sendText(JSONUtil.toJsonStr(text));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 此为单点消息(多人)
     *
     * @param websocketVo webSocket服务端到客户端vo
     */
    public void sendMoreMessage(WebsocketVo websocketVo) {
        List<String> acceptIds = websocketVo.getAcceptIds();
        if (CollectionUtils.isEmpty(acceptIds)) {
            return;
        }
        acceptIds.forEach(acceptId -> {
            Session session = sessionPool.get(acceptId);
            if (session != null && session.isOpen()) {
                try {
                    log.info("【websocket消息】 单点消息:" + websocketVo);
                    session.getAsyncRemote().sendText(JSONUtil.toJsonStr(websocketVo));
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }

    /**
     * 查看当前登录用户是否有未读消息，如果有则进行推送
     *
     * @param userId 用户id
     */
    public void offlineNews(String userId) {
        List<WebsocketVo> websocketVos = webSocket.newsService.offlineNews(userId);
        if (CollectionUtil.isNotEmpty(websocketVos)) {
            webSocket.sendOneNewsMessage(JSONUtil.toJsonStr(websocketVos), userId);
        }
    }

    /**
     * 查看当前登录用户是否有未读通知，如果有则进行推送
     *
     * @param userId 用户id
     */
    public void offlineNotice(String userId) {
        List<WebsocketVo> websocketVos = webSocket.noticeService.offlineNotice(userId);
        if (CollectionUtil.isNotEmpty(websocketVos)) {
            webSocket.sendOneNewsMessage(JSONUtil.toJsonStr(websocketVos), userId);
        }
    }
}
